package com.shujia.trip.ods;

import com.shujia.trip.utils.FileReaderUtil;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class FlinkSqlRunTool {
    public static void main(String[] args) {
        if (args.length == 0) {
            return;
        }
        String path = args[0];

        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                // 指定为流处理的运行模式
                .inStreamingMode()
                // 指定为批处理的运行模式
                //.inBatchMode()
                .build();


        // 第二种方式构建Table环境
        TableEnvironment tEnv = TableEnvironment.create(settings);

        System.out.println("=========================================================");
        System.out.println("正在处理的CDC SQL脚本：" + path);
        System.out.println("=========================================================");

        // 加载Flink CDC采集脚本
        String sqlStr = FileReaderUtil.readFile(path);
        // 提取脚本名称
        String[] splits = path.split("/");
        String jobName = splits[splits.length - 1];

        // 切分SQL 循环执行每一条SQL语句
        for (String sql : sqlStr.split(";")) {
            if (!sql.trim().equals("")) {
//                System.out.println(sql);
                tEnv.executeSql(sql);

                // 在代码中通过JDBC保存jobId
//                TableResult tableResult = tEnv.executeSql(sql + ";");
//
//                // 如果最终触发了任务的提交则会获得一个jobClient
//                Optional<JobClient> jobClient = tableResult.getJobClient();
//                // 判断是否有jobClient
//                if (jobClient.isPresent()) {
//                    String jobId = jobClient.get().getJobID().toString();
//                    // 建立连接
//                    Connection conn = DriverManager.getConnection("jdbc:mysql://master:3306/flink", "root", "123456");
//                    // 构建statement
//                    PreparedStatement pSt = conn.prepareStatement("replace into task_job_id values(?,?)");
//
//                    // 设置参数
//                    pSt.setString(1, jobName);
//                    pSt.setString(2, jobId);
//
//                    // 执行SQL
//                    pSt.execute();
//
//                    // 关闭连接
//                    pSt.close();
//                    conn.close();
//            }
            }
        }
    }
}
